
package mvcc

import (
	"context"

	"go.etcd.io/etcd/api/v3/mvccpb"
	"go.etcd.io/etcd/pkg/v3/traceutil"
	"go.etcd.io/etcd/server/v3/lease"
	"go.etcd.io/etcd/server/v3/storage/backend"
	"go.etcd.io/etcd/server/v3/storage/schema"
	"go.uber.org/zap"
)

type storeTxnRead struct {
	s  *store	// 该storeTxnrRead实例关联的store实例。
	tx backend.ReadTx	// 该storeTxnrRead实例关联的ReadTx实例。
	firstRev int64	// 对应于关联的store实例的firstRev和rev宇段。
	rev      int64
	trace *traceutil.Trace
}

func (s *store) Read(mode ReadTxMode, trace *traceutil.Trace) TxnRead {
	s.mu.RLock()
	s.revMu.RLock()
	// 对于只读工作负载，我们通过复制事务读取缓冲区来使用共享缓冲区
	// 以提高正在进行的阻塞写入的并发性。
	// 对于写/读事务，我们使用共享缓冲区
	// 而不是复制事务读取缓冲区，以避免事务开销。
	var tx backend.ReadTx
	if mode == ConcurrentReadTxMode {
		tx = s.b.ConcurrentReadTx()
	} else {
		tx = s.b.ReadTx()
	}

	tx.RLock() // RLock为no-op。创建concurrentReadTx后不需要锁定它。
	firstRev, rev := s.compactMainRev, s.currentRev
	s.revMu.RUnlock()
	return newMetricsTxnRead(&storeTxnRead{s, tx, firstRev, rev, trace})
}

func (tr *storeTxnRead) FirstRev() int64 { return tr.firstRev }
func (tr *storeTxnRead) Rev() int64      { return tr.rev }

func (tr *storeTxnRead) Range(ctx context.Context, key, end []byte, ro RangeOptions) (r *RangeResult, err error) {
	return tr.rangeKeys(ctx, key, end, tr.Rev(), ro)
}

// storeTxnRead.End（）方法中，
// 会释放在store.Read（）方法中获取的store.mu上的读锁和底层只读事务上的锁
func (tr *storeTxnRead) End() {
	tr.tx.RUnlock() // 运行锁发出concurrentReadTx结束的信号。
	tr.s.mu.RUnlock()
}

// storeTxnWrite实现了TxnWrite接口，其中内嵌了storeTxnRead
type storeTxnWrite struct {
	storeTxnRead
	tx backend.BatchTx	// 当前storeTxnWrite实例关联的读写事务。
	// 记录创建当前storeTxnWrite实例时store.currentRev字段的值。
	beginRev int64
	// 在当前读写事务中发生改动的键值对信息。
	changes  []mvccpb.KeyValue
}

func (s *store) Write(trace *traceutil.Trace) TxnWrite {
	s.mu.RLock()	// 加读锁
	tx := s.b.BatchTx()	// 获取读写事务
	tx.Lock()	// 获取读写事务的锁
	tw := &storeTxnWrite{	// 创建storeTxnWrite实例
		storeTxnRead: storeTxnRead{s, tx, 0, 0, trace},	//其中firstRev初始化为0
		tx:           tx,
		beginRev:     s.currentRev,
		changes:      make([]mvccpb.KeyValue, 0, 4),
	}
	return newMetricsTxnWrite(tw)
}

func (tw *storeTxnWrite) Rev() int64 { return tw.beginRev }

// storeTxn Write.Range（）方法也是调用其rangeKeys（）方法实现的，
// 与stor巳TxnRead.Range（）方法的主要区别在于传入的rev参数
func (tw *storeTxnWrite) Range(ctx context.Context, key, end []byte, ro RangeOptions) (r *RangeResult, err error) {
	rev := tw.beginRev
	if len(tw.changes) > 0 {
		// 如采当前读写事务中有关新键位对的操作，则该方法也需要能查询到这些更新的键值对，
		// 所以传入的rev参数是beginRev+l
		rev++
	}
	return tw.rangeKeys(ctx, key, end, rev, ro)
}

// storeTxn Write.DeleteRange（）方法是通过调用deleteRange（）方法实现
func (tw *storeTxnWrite) DeleteRange(key, end []byte) (int64, int64) {
	if n := tw.deleteRange(key, end); n != 0 || len(tw.changes) > 0 {
		return n, tw.beginRev + 1	// 根据当前事务是否有更新剧字，决定返回的main revision值
	}
	return 0, tw.beginRev
}

// storeTxnWrite.Put（）方法的实现，该方法实现了向存储中追加一个键值对数据的功能，
// 其底层是通过调用storeTxnWrite.put（）方法实现的
func (tw *storeTxnWrite) Put(key, value []byte, lease lease.LeaseID) int64 {
	tw.put(key, value, lease)
	return tw.beginRev + 1
}

//End（）方法，其中主要完成三个操作，
// 一是递增store.currentRev，
// 二是将Consistentlndex记录到名为“me旷的Bucket中，
// 三是调用Unlock()方法提交当前事务并开启新的读写事务。
func (tw *storeTxnWrite) End() {
	// 检测当前读写事务中是否有修改操作
	if len(tw.changes) != 0 {
		// 修改store中的currentRev字段，加读锁同步
		tw.s.revMu.Lock()
		// 递增currentRev
		tw.s.currentRev++
	}
	// batchTxBuffered及batchTx中的Unlock（）方法实现，
	// 其中会将当前读写事务提交,同时开启新的读写事务
	tw.tx.Unlock()
	if len(tw.changes) != 0 {
		// 修改currentRev完成，释放读锁
		tw.s.revMu.Unlock()
	}
	tw.s.mu.RUnlock()	// 释放读锁
}

// 该方法的大致流程是：首先扫描内存索引（即前面介绍的treelndex），
// 得到对应的revision，然后通过r巳vision查询BoltDB得到真正的键值对数据，
// 最后将键值对数据反序列化成KeyValue 并封装成Rang巳Result实例返回。
func (tr *storeTxnRead) rangeKeys(ctx context.Context, key, end []byte, curRev int64, ro RangeOptions) (*RangeResult, error) {
	rev := ro.Rev
	// 检测RangeOptions中指定的revision信息是否合法，如果不合法，
	// 则根据curRev（对storeTxnRead来说，就是rev字段）当前的compactMainRev值对rev进行校正
	if rev > curRev {
		return &RangeResult{KVs: nil, Count: -1, Rev: curRev}, ErrFutureRev
	}
	if rev <= 0 {
		rev = curRev
	}
	if rev < tr.s.compactMainRev {
		return &RangeResult{KVs: nil, Count: -1, Rev: 0}, ErrCompacted
	}
	// 只获取键位对的个数，不需妥返回具体的键位对数据，所以只查询索引即可
	if ro.Count {
		total := tr.s.kvindex.CountRevisions(key, end, rev)
		tr.trace.Step("count revisions from in-memory index tree")
		return &RangeResult{KVs: nil, Count: total, Rev: curRev}, nil
	}
	revpairs, total := tr.s.kvindex.Revisions(key, end, rev, int(ro.Limit))
	tr.trace.Step("range keys from in-memory index tree")
	// 根据RangeOptions中指定的参数，创建RangeResult实例作为返回值
	if len(revpairs) == 0 {
		return &RangeResult{KVs: nil, Count: total, Rev: curRev}, nil
	}

	limit := int(ro.Limit)
	if limit <= 0 || limit > len(revpairs) {
		limit = len(revpairs)
	}

	kvs := make([]mvccpb.KeyValue, limit)
	revBytes := newRevBytes()
	for i, revpair := range revpairs[:len(kvs)] {
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		default:
		}
		revToBytes(revpair, revBytes)
		_, vs := tr.tx.UnsafeRange(schema.Key, revBytes, nil, 0)
		if len(vs) != 1 {
			tr.s.lg.Fatal(
				"range failed to find revision pair",
				zap.Int64("revision-main", revpair.main),
				zap.Int64("revision-sub", revpair.sub),
			)
		}
		if err := kvs[i].Unmarshal(vs[0]); err != nil {
			tr.s.lg.Fatal(
				"failed to unmarshal mvccpb.KeyValue",
				zap.Error(err),
			)
		}
	}
	tr.trace.Step("range keys from bolt db")
	// 将kvs封装成RangeResult实例并返回
	return &RangeResult{KVs: kvs, Count: total, Rev: curRev}, nil
}

// storeTxnWrite.Put（） 方法， 它在执行更新操作之前会先将 Key 值封装成 Leaseltem 实例 ， 之
// 后通过 lessor. GetLease （） 方法查找对应的 Lease 实例， 然后完成对键值对的更新操作，最后将更
// 新的键值对与原来的 Lease 实例解绑，并与新指定的 Leas e 实例绑定。
func (tw *storeTxnWrite) put(key, value []byte, leaseID lease.LeaseID) {
	rev := tw.beginRev + 1	// 当前事务产生的修改对应的mainrevision部分都是该值
	c := rev
	oldLease := lease.NoLease

	// 在内存索引中查找对应的键值对信息，在后面创建KeyValue实例时会使用到这些值
	_, created, ver, err := tw.s.kvindex.Get(key, rev)
	if err == nil {	// 记录键位对绑定的 Lease 实例
		c = created.main
		oldLease = tw.s.le.GetLease(lease.LeaseItem{Key: string(key)})
		tw.trace.Step("get key's previous created_revision and leaseID")
	}
	ibytes := newRevBytes()
	// 创建此次Put操作对应的revision实例，其中mainrevision部分为beginRev+ 1
	idxRev := revision{main: rev, sub: int64(len(tw.changes))}
	// 将mainreviSion和subrevision两部分写入ibytes中，后续会作为写入BoltDB的Key
	revToBytes(idxRev, ibytes)

	ver = ver + 1
	kv := mvccpb.KeyValue{	// 创建KeyValue实例
		Key:            key,	// 原始Keyf直
		Value:          value,	// 原始Value值
		// 如果内存索引中已经存在该键值对，则创建版本不变，
		// 否则为新建Key，贝1J将CreateRevision设直成beginRev+ 1
		CreateRevision: c,
		ModRevision:    rev,	// beginRev + 1
		Version:        ver,	// 递增verison
		Lease:          int64(leaseID),
	}

	d, err := kv.Marshal()	// 将上面创建的KeyValue实例序列化，
	if err != nil {
		tw.storeTxnRead.s.lg.Fatal(
			"failed to marshal mvccpb.KeyValue",
			zap.Error(err),
		)
	}

	tw.trace.Step("marshal mvccpb.KeyValue")
	// 将Key(mainrevision和subrevision组成）和Value（上述KeyValue实例序列化的结采）
	// 写入II BoltDB中名为“key”的Bucket
	tw.tx.UnsafeSeqPut(schema.Key, ibytes, d)
	tw.s.kvindex.Put(key, idxRev)	// 	将原始Key与revision实例的对应关系写入内存索引中
	tw.changes = append(tw.changes, kv)	// 将上述KeyValue写入到changes中
	tw.trace.Step("store kv pair into bolt db")

	if oldLease != lease.NoLease {	// 将键位对与 之前的 Lease 实例解绑
		if tw.s.le == nil {
			panic("no lessor to detach lease")
		}
		err = tw.s.le.Detach(oldLease, []lease.LeaseItem{{Key: string(key)}})
		if err != nil {
			tw.storeTxnRead.s.lg.Error(
				"failed to detach old lease from a key",
				zap.Error(err),
			)
		}
	}
	if leaseID != lease.NoLease {	// 将键位对与新指定的 Lease 实例绑定
		if tw.s.le == nil {
			panic("no lessor to attach lease")
		}
		err = tw.s.le.Attach(leaseID, []lease.LeaseItem{{Key: string(key)}})
		if err != nil {
			panic("unexpected error from lease Attach")
		}
	}
	tw.trace.Step("attach lease to kv pair")
}

// storeTxnWrite.deleteRange（）方法中，首先会从内存索引查询待删除的Key和对应的revision实例，
// 然后会调用storeTxnWrite.delete（）方法逐个删除
func (tw *storeTxnWrite) deleteRange(key, end []byte) int64 {
	rrev := tw.beginRev
	if len(tw.changes) > 0 {	// 根据当前事务是否有更新操作，决定后续使用的main revision值
		rrev++
	}
	// 在内存索引中查询待删除的key
	keys, _ := tw.s.kvindex.Range(key, end, rrev)
	if len(keys) == 0 {
		return 0
	}
	for _, key := range keys {	// 遍历待删除的key，并调用delete（）方法完成删除
		tw.delete(key)
	}
	return int64(len(keys))	// 返回删除的key值
}

// storeTxnWrite.delete（）方法中会向BoltDB和内存索引中添加tombstone键值对，
// storeTxnWrite.Delete （） 方法中 除了添加 Tombstone ，还会解绑键值对与 Lease 实例
func (tw *storeTxnWrite) delete(key []byte) {
	// 将revision转换成BoltDB中的key
	ibytes := newRevBytes()
	idxRev := revision{main: tw.beginRev + 1, sub: int64(len(tw.changes))}
	revToBytes(idxRev, ibytes)
	// 追加一个't’标识Tombstone
	ibytes = appendMarkTombstone(tw.storeTxnRead.s.lg, ibytes)
	// 创建KeyValue实例，其中只包含key，并进行序列化
	kv := mvccpb.KeyValue{Key: key}
	d, err := kv.Marshal()
	if err != nil {
		tw.storeTxnRead.s.lg.Fatal(
			"failed to marshal mvccpb.KeyValue",
			zap.Error(err),
		)
	}
	// 将上面生成的key(ibytes）和value(KeyValue序列化后的数据）写入名为“key”的Bucket中
	tw.tx.UnsafeSeqPut(schema.Key, ibytes, d)
	// 在内存索引中添加Tombstone
	err = tw.s.kvindex.Tombstone(key, idxRev)
	if err != nil {
		tw.storeTxnRead.s.lg.Fatal(
			"failed to tombstone an existing key",
			zap.String("key", string(key)),
			zap.Error(err),
		)
	}
	// 向changes中追加上述KeyValue实例
	tw.changes = append(tw.changes, kv)

	item := lease.LeaseItem{Key: string(key)}	// 创建 LeaseItem 实例
	leaseID := tw.s.le.GetLease(item)	// 根据 LeaseItem 查找对应的 Lease 实例

	if leaseID != lease.NoLease {
		err = tw.s.le.Detach(leaseID, []lease.LeaseItem{item})	// 调用 Detach （）方法解绑
		if err != nil {
			tw.storeTxnRead.s.lg.Error(
				"failed to detach old lease from a key",
				zap.Error(err),
			)
		}
	}
}

// storeTxn Write.Changes（）方法直接返回其changes字段
func (tw *storeTxnWrite) Changes() []mvccpb.KeyValue { return tw.changes }
